package yxy.cherry.game.base.message;

import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.google.protobuf.ByteString;

import yxy.apple.protobuf.Constants;
import yxy.apple.protobuf.Message.Parcal;

public class MessageStation {

	private static class Message {
		private int sequence = 0;
		private String tag = null;
		private long source = 0;
		private long destination = 0;
		private int proto = Constants.ControlProto.Unknown;
		private ByteString data = null;
		private boolean terminus = false;

		private static Message convert(Parcal parcal) {
			Message instance = new Message();
			instance.sequence = parcal.getSequence();
			instance.tag = parcal.getTag();
			instance.source = parcal.getSource();
			instance.destination = parcal.getDestination();
			instance.proto = parcal.getProto();
			instance.data = parcal.getData();
			instance.terminus = false;
			return instance;
		}

		private static Message terminus() {
			Message instance = new Message();
			instance.terminus = true;
			return instance;
		}

		private Message() {
		}

	}

	public abstract class Consumer implements Runnable {
		protected abstract void consume(int sequence, String tag, long source, long destination, int proto, ByteString data);

		public void run() {
			MessageStation.this.treating++;
//			System.err.println("Consumer run::" + MessageStation.this.treating);
			while (MessageStation.this.running && !Thread.interrupted()) {
				try {
					Message message = MessageStation.this.queue.take();
//					System.err.println("queue size::" + MessageStation.this.overstocked());
					if (message != null) {
						if (message.terminus) {
							break;// 退出线程
						} else {
							this.consume(message.sequence, message.tag, message.source, message.destination, message.proto, message.data);
						}
					}
				} catch (Exception e) {
				}
			}
			MessageStation.this.treating--;
		}
	}

	private int treating = 0;
	private boolean running = false;

	public LinkedTransferQueue<Message> queue = new LinkedTransferQueue<>();

	private ThreadPoolExecutor executor;

	public MessageStation() {

	}

	/**
	 * 当前正在接待的消费者数量
	 * 
	 * @return
	 */
	public int treating() {
		return this.treating;
	}

	/**
	 * 接待消费者，每个消费者会开启一个处理线程
	 * 
	 * @param consumer
	 */
	public void treat(Consumer consumer) {
		if (this.running) {
			this.executor.execute(consumer);
		}
	}

	/**
	 * 减少n个消费者
	 * 
	 * @param count
	 */
	public void reduce(int n) {
		for (int i = 0; i < n; i++) {
			this.deliver(Message.terminus());
		}
	}

	/**
	 * 开启消息站进行服务
	 */
	public void open(int threads) {
		// TODO 用config配置线程池参数、配置线程工厂和饱和策略
		this.executor = new ThreadPoolExecutor(threads, threads, 0, TimeUnit.SECONDS, new LinkedTransferQueue<Runnable>());
		this.executor.prestartAllCoreThreads();
		this.running = true;
	}

	/**
	 * 关闭消息站
	 */
	public void close() {
		this.running = false;
		this.executor.shutdown();
		while (this.queue.hasWaitingConsumer()) {
			this.deliver(Message.terminus());
		}
	}

	public boolean running() {
		return this.running;
	}

	/**
	 * 队列积压消息数量
	 * 
	 * @return
	 */
	public int overstocked() {
		return this.queue.size();
	}

	public void deliver(Parcal parcal) {
		this.deliver(Message.convert(parcal));

	}

	private void deliver(Message message) {
		if (!this.queue.tryTransfer(message)) {
			this.queue.offer(message);
		}
	}
}
